package com.bjb.vr.client.rabbitmq;

import cn.hutool.core.lang.UUID;
import cn.hutool.crypto.digest.HMac;
import cn.hutool.crypto.digest.HmacAlgorithm;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bjb.vr.client.config.CanalConfig;
import com.bjb.vr.client.properties.ServerInfoProperties;
import com.bjb.vr.client.utils.HttpSender;
import com.bjb.vr.common.constant.CanalConstant;
import com.bjb.vr.common.constant.EncryptConstant;
import com.bjb.vr.common.constant.ServerModeConstant;
import com.bjb.vr.common.dto.ExecutedSql;
import com.bjb.vr.common.exception.CommonException;
import com.bjb.vr.common.service.ExecutedSqlService;
import com.bjb.vr.common.utils.CanalTool;
import com.bjb.vr.common.utils.DateTimeUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.HashMap;
import java.util.Objects;

/**
 * @version 1.0.0
 * @author: HuJingBo
 * @time: 2022/6/24 14:30
 */
@Slf4j
@Component
public class ClientDataSyncListener {

    @Resource
    private ServerInfoProperties serverInfoProperties;

    @Resource
    private JdbcTemplate jdbcTemplate;

    @Resource
    private ExecutedSqlService executedSqlService;

    /**
     * 监听RabbitMQ消息队列
     * <p>
     * 同步数据说明: 下面的数据同步的代码，其实是给混合云部署模式的用户使用的，私有化部署和云端部署，是不需要进行数据同步的
     * 1、如果客户买的是私有化部署模式，不需要进行同步操作，操作自己本地数据库即可
     * 2、如果客户买的是云部署模式，则不需要同步，用同一套数据库即可
     * 3、如果客户买的是混合云模式，则如果在同步表中存在，并且配置文件开启了数据同步，则进行同步操作；混合云部署模式中需要有一个本地部署服务和一个云端服务
     * <p>
     * 注意事项：
     * 1、配置服务的时候，首先需要确认服务的类型:是私有化、云端、混合云
     * 2、如果是私有化，需要将配置文件改成不同步数据
     * 3、如果是云端，不需要修改配置文件，不把租户ID添加到sync_tenant表中即可
     * 4、如果是混合云，需要同步数据时，将私有化的配置文件开启数据同步，然后将租户的ID保存到云端sync_tenant表中即可
     * <p>
     * 数据回环问题解决：
     * 1、描述：假设在A库插入数据，A库产生binlog，之后同步到B库，B库同样也会产生binlog。由于是双向同步，这条记录，又会被重新同步回A库。由于A库应存在这条记录了，产生冲突
     * 2、解决方式：将执行过的SQL保存到executed_sql表中，每次执行操作时查询数据，存在则跳过，并将数据清理掉
     *
     * @param message
     */
    @RabbitListener(queues = {CanalConfig.QUEUE_INFO_CANAL})
    public void receiveCanal(Message message, Channel channel) throws IOException {
        try {
            // 不需要同步
            if (!serverInfoProperties.getNeedSyncDatabase()) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }

            JSONObject object = JSONObject.parseObject(new String(message.getBody()));
            String eventType = object.getString(CanalConstant.EVENT_TYPE);
            String statement = CanalTool.parseSqlHandle(object);
            // SQL为空,直接确认
            if (StringUtils.isBlank(statement)) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }

            try {
                // 同步数据库到远程服务: 本地-->云端
                if (StringUtils.isNotBlank(statement) && ServerModeConstant.LOCAL.equals(serverInfoProperties.getDeployMode())) {
                    sendDataToServer(eventType, statement, object);
                }
                // 当上面的方法return后,说明不符合，在这里进行确认操作
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception e) {
                log.error(e.getMessage());
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            }
        } catch (Exception e) {
            // 报错时确认消息，否则可能造成rabbitmq消息堆满
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            throw new CommonException(e.getMessage());
        }
    }


    /**
     * 发送同步数据给云端
     *
     * @param executeType
     * @param statement
     * @param object
     */
    public void sendDataToServer(String executeType, String statement, JSONObject object) throws Exception {
        String tenantId = ignoreRules(object);
        if (StringUtils.isBlank(tenantId)) {
            return;
        }

        // 判断配置文件中租户ID和SQL中租户ID是否一致
        if (!tenantId.equals(serverInfoProperties.getTenantId())) {
            return;
        }

        // 首先查询一下executed_sql表，检查有没有执行过该SQL，如果有删除即可。
        ExecutedSql executedEntity = CanalTool.buildExecuteEntity(object, statement, tenantId);
        ExecutedSql executedSql = executedSqlService.queryByMd5(executedEntity.getMd5(), tenantId);
        if (Objects.nonNull(executedSql)) {
            log.info("已经执行过该SQL,不再重复执行,SQL语句MD5值: {}", executedEntity.getMd5());
            // 执行删除操作
            executedSqlService.deleteByMd5(executedEntity.getMd5(), tenantId);
            return;
        }

        try {
            // 保存数据到executed_sql中
            executedSqlService.saveDataToExecute(executedEntity);
        } catch (Exception e) {
        }

        // 生成签名
        StringBuilder builder = new StringBuilder();
        String requestId = UUID.fastUUID().toString();
        String requestTime = DateTimeUtil.nowStrTime();
        String data = builder.append(tenantId).append(requestId).append(requestTime).toString();
        byte[] key = EncryptConstant.ENCRYPT_KEY.getBytes();
        HMac mac = new HMac(HmacAlgorithm.HmacMD5, key);
        String signed = mac.digestHex(data);

        HashMap<String, String> params = new HashMap<>();
        params.put("dbName", object.getString(CanalConstant.DB));
        params.put("tableName", object.getString(CanalConstant.TABLE));
        params.put("type", executeType);
        params.put("tenantId", tenantId);
        params.put("requestId", requestId);
        params.put("requestTime", requestTime);
        params.put("statement", statement);
        params.put("signature", signed);

        // 发送http请求
        HttpSender.postSync(serverInfoProperties.getSendSyncDataUrl(), params);
    }

    /**
     * 忽略不包含租户ID的同步数据
     * 忽略不需要同步的租户数据
     *
     * @param object
     * @return
     */
    private String ignoreRules(JSONObject object) {
        JSONArray dataArray = (JSONArray) object.get(CanalConstant.DATA);
        JSONObject jsonObject = dataArray.getJSONObject(0);
        // 1、不同步没有租户ID字段的数据
        boolean isReal = jsonObject.containsKey(CanalConstant.TENANT_ID);
        if (!isReal) {
            return null;
        }

        // 2、当服务类型为local时，需要指定TenantId，否则不同步；
        String tenantId = serverInfoProperties.getTenantId();
        if (StringUtils.isBlank(tenantId) && ServerModeConstant.LOCAL.equals(serverInfoProperties.getDeployMode())) {
            log.error("请在配置文件中配置租户的ID...");
            return null;
        }
        return jsonObject.getString(CanalConstant.TENANT_ID);
    }

}
